package com.giraffe.mqtt.handler;

import com.alibaba.fastjson2.JSON;
import com.giraffe.cloud.socket.annotation.TcpEndpointHandler;
import com.giraffe.cloud.socket.mapping.AbstractTcpEndpointHandler;
import com.giraffe.cloud.socket.pojo.TcpSession;
import com.giraffe.mqtt.service.ChannelService;
import com.giraffe.mqtt.service.MqttHandlerService;
import com.giraffe.mqtt.service.MqttService;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @Author ldx
 * @Date 2024/01/15 09:47
 **/
@Slf4j
@TcpEndpointHandler
@Component
public class MqttHandler extends AbstractTcpEndpointHandler {

    @Resource
    MqttHandlerService mqttHandlerService;
    @Resource
    ChannelService channelService;
    
    @Override
    public void doOnOpen(TcpSession ctx) {
        System.out.println("---------1----------");
    }

    @Override
    public void doOnMessage(TcpSession ctx, Object msg) {
        System.out.println("---------2----------");
        log.info(JSON.toJSONString(msg));
        MqttMessage m = (MqttMessage) msg;

        MqttMessageType mqttMessageType = m.fixedHeader().messageType();
        switch (mqttMessageType){
            case CONNECT:
                log.info("CONNECT");
                mqttHandlerService.login(ctx.channel(), (MqttConnectMessage)m);
                break;
            case PUBLISH:
                log.info("PUBLISH");
                mqttHandlerService.publish(ctx.channel(), (MqttPublishMessage)m);
                break;
            case SUBSCRIBE:
                log.info("SUBSCRIBE");
                mqttHandlerService.subscribe(ctx.channel(), (MqttSubscribeMessage)m);
                break;
            case UNSUBSCRIBE:
                log.info("UNSUBSCRIBE");
                mqttHandlerService.unsubscribe(ctx.channel(),(MqttUnsubscribeMessage)m);
                break;
            case PINGREQ:
                log.info("PINGREQ");
                mqttHandlerService.pong(ctx.channel());
                break;
            case DISCONNECT:
                log.info("DISCONNECT");
                mqttHandlerService.disconnect(ctx.channel());
                break;
            case PUBACK:
                log.info("PUBACK");
                mqttHandlerService.puback(ctx.channel(),m);
                break;
            case PUBREC:
                log.info("PUBREC");
                mqttHandlerService.pubrec(ctx.channel(),m);
                break;
            case PUBREL:
                log.info("PUBREL");
                mqttHandlerService.pubrel(ctx.channel(),m);
                break;
            case PUBCOMP:
                log.info("PUBCOMP");
                mqttHandlerService.pubcomp(ctx.channel(),m);
                break;
            default:
                break;
        }

    }

    @Override
    public void doOnError(TcpSession ctx, Throwable cause) {
        System.out.println("---------3----------");
    }

    @Override
    public void doOnClose(TcpSession ctx) {
        System.out.println("---------4----------");
        String deviceId = channelService.getDeviceId(ctx.channel());
        log.info("设备下线:{}",deviceId);
        mqttHandlerService.close(ctx.channel());
    }

    @Override
    public void doOnEvent(TcpSession ctx, Object evt) {
        System.out.println("---------5----------");
    }
}
